package com.changan.examples;

import com.changan.utils.FlinkRestApiUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.*;

public class StopWithSavepoint {
    private static final String FLINK_HOST = "";
    private static final String FLINK_PORT = "";

    @SneakyThrows
    public static void main(String[] args) throws JsonProcessingException {
        ObjectMapper objectMapper = new ObjectMapper();

        String flinkJob = FlinkRestApiUtil.getListOfFlinkJob(FLINK_HOST, FLINK_PORT);
        System.out.println(flinkJob);

        JsonNode jsonNode = objectMapper.readTree(flinkJob);
        String jobId = "";
        for (JsonNode job : jsonNode.get("jobs")) {
            if (job.get("status").asText().equals("RUNNING")) {
                jobId = job.get("id").asText();
            }
        }
        System.out.println(jobId);
        StopArgs stopArgs = new StopArgs();
        stopArgs.setDrain(false);
        stopArgs.setTargetDirectory("hdfs://nameservice1/tmp/yarnjobs/");
        String response = FlinkRestApiUtil.stopFlinkJobWithSavepoint(
                FLINK_HOST,
                FLINK_PORT,
                jobId,
                stopArgs
        );
        //{"request-id":"d235b264d99618ca3714c812c47525a8"}
        System.out.println(response);
        JsonNode requestIdNode = objectMapper.readTree(response.toString());
        String requestId = requestIdNode.get("request-id").asText();
        String statusOfSaOp = "";
        for (int i = 0; i < 5; i++) {
            System.out.println("这是第" + i + "次查询savapoint是否成功.");
            statusOfSaOp = FlinkRestApiUtil.getStatusOfSaOp(FLINK_HOST, FLINK_PORT, jobId, requestId);
            if (objectMapper.readTree(statusOfSaOp).get("status").get("id").asText().equals("COMPLETED")) {
                System.out.println(statusOfSaOp);
                //break;
            }
            Thread.sleep(1000);
        }


    }

    @Getter
    @Setter
    @NoArgsConstructor
    @AllArgsConstructor
    public static class StopArgs {
        private Boolean drain;
        private String targetDirectory;
    }

}
